package com.bw.yk08;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;

public class MyKafkaUtil {

    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String topic,String groupid)
    {
        Properties pro=new Properties();
        pro.setProperty("bootstrap.servers", "hadoop-single:9092");
        pro.setProperty("group.id", groupid);

        FlinkKafkaConsumer<String> flinkKafkaConsumer=new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),pro);
        flinkKafkaConsumer.setStartFromEarliest();

        return flinkKafkaConsumer;
    }

    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic)
    {
        Properties pro=new Properties();
        pro.setProperty("bootstrap.servers", "hadoop-single:9092");
        return new FlinkKafkaProducer<String>(topic,new SimpleStringSchema(),pro);
    }
}
